package com.xzx.flink.streamapi.state;

import com.xzx.flink.bean.ClickEvent;
import com.xzx.flink.streamapi.source.ClickSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;

/**
 * 模拟滚动窗口，10秒计算一次url的pv数据
 *
 * @version 1.0
 * @auther xinzhixuan
 * @date 2022/5/14 00:59
 */
public class MapState_01_UrlPV {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env.addSource(new ClickSource()).assignTimestampsAndWatermarks(WatermarkStrategy.<ClickEvent>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<ClickEvent>() {
                    @Override
                    public long extractTimestamp(ClickEvent element, long recordTimestamp) {
                        return element.getTimestamp();
                    }
                })).keyBy(ClickEvent::getUrl)
                //10秒窗口的url pv
                .process(new MapStateWindowProcessFunction(10000L)).print();

        env.execute(MapState_01_UrlPV.class.getSimpleName());
    }

    public static class MapStateWindowProcessFunction extends KeyedProcessFunction<String, ClickEvent, String> {

        private final long windowSize;
        /**
         * key：窗口的开始时间
         * value：窗口的pv数
         */
        private MapState<Long, Long> state;

        public MapStateWindowProcessFunction(long windowSize) {
            this.windowSize = windowSize;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            state = getRuntimeContext().getMapState(new MapStateDescriptor<>("pv", Long.class, Long.class));
        }

        @Override
        public void processElement(ClickEvent value, KeyedProcessFunction<String, ClickEvent, String>.Context ctx, Collector<String> out) throws Exception {
            //1.当前事件属于哪个窗口
            long windowStart = value.getTimestamp() / windowSize * windowSize;
            Long pv = state.get(windowStart);
            if (pv == null) {
                state.put(windowStart, 1L);
            } else {
                state.put(windowStart, pv + 1);
            }
            // 注册一个10秒后的事件时间,这里减一的原因是，这里是事件时间，时间的推进靠的是水位线，也就是水位线推进到窗口的结束时间时，这个窗口就关闭,表示这个窗口内的所有数据已经到齐
            ctx.timerService().registerEventTimeTimer(windowStart + windowSize - 1);
        }

        @Override
        public void onTimer(long timestamp, KeyedProcessFunction<String, ClickEvent, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
            // 有一个10秒的窗口要关闭了
            long windowStart = timestamp + 1 - windowSize;
            Long pv = state.get(windowStart);
            out.collect("url:" + ctx.getCurrentKey() + "\tpv=" + pv.toString() +"\t窗口时间：" + new Timestamp(windowStart) + "~" + new Timestamp(timestamp + 1));
            //从状态中清楚数据
            state.remove(windowStart);
        }
    }
}
